package com.ddxz.flink

import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.io.jdbc.{JDBCInputFormat, JDBCOutputFormat}
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.types.Row

object Jdbc {

  def getRow(r1: Row) :String = {
    val rst: StringBuffer = new StringBuffer()
    for(item <- 0 to r1.getArity) {
      rst.append(item).append(",")
    }

    rst.toString
  }

  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val typName = Array("id", "user", "status", "createTime", "realName", "schema")
    val types = new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO,
      BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
    val t2 = new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
    val t2Name = Array("id", "tableId", "realName")
    val ds1 = JDBCInputFormat.buildJDBCInputFormat()
      .setDBUrl("jdbc:mysql://192.168.12.140:3306/demo?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
      .setDrivername("com.mysql.cj.jdbc.Driver")
      .setUsername("root")
      .setPassword("root")
      .setQuery("SELECT c_id, c_user, c_status, c_create_time, c_real_name, c_schema from t_meta_object")
      .setRowTypeInfo(types)
 //     .setRowTypeInfo(types)
      .finish()
    ds1.openInputFormat()

    val ds2 = JDBCInputFormat.buildJDBCInputFormat()
      .setDBUrl("jdbc:mysql://192.168.12.140:3306/demo?useUnicode=true&characterEncoding=utf8&autoReconnect=true&failOverReadOnly=false&serverTimezone=Asia/Shanghai")
      .setDrivername("com.mysql.cj.jdbc.Driver")
      .setUsername("root")
      .setPassword("root")
      .setQuery("select c_id, c_rid, c_real_name, c_display_name from t_meta_column")
      .setRowTypeInfo(t2)
      .finish()
    ds2.openInputFormat()

    var d1 = env.createInput(ds1)
    var d2 = env.createInput(ds2)
//    d1.partitionByHash(0).join(d2.partitionByHash(1))
//      .where(0)
//      .equalTo(1)
//      .apply((t1, t2) => (t1.getField(3), t1.getField(4), t1.getField(5),
//        t2.getField(2), t2.getField(3)))
//      .print()
//    var rst :JSONObject = _
    d2.filter{_.getField(1) == null }.count()
    d2.filter{_.getField(1) != null }.count()
// count   d2.filter{_.getField(1) == null }.flatMap(dt => rst.put(dt.getArity.toString, 1))
    d1.join(d2)
      .where(0)
      .equalTo(1)
//      .apply((t1, t2) => (t1._, t2._))
      //      .apply(DdxzJoin)
      .apply((t1, t2) => (t1.getField(3), t1.getField(4), t1.getField(5),
        t2.getField(2), t2.getField(3)))
      //      .sortPartition(1, Order.DESCENDING)
      .print()
    val num = d1.count()
    val num2 = d2.count()
    d1.first(3).map(r1 => (r1, num, num+num2)).print()
  }
}
